package com.feeyo.prophet.plugins;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.ObjectUtils;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

import com.feeyo.llm.agent.engine.ProcessEnum;
import com.feeyo.llm.util.JsonUtil;
import com.feeyo.prophet.exception.InterfaceQueryResult;
import com.feeyo.prophet.pojo.action.flight.FlightDynamicReq;
import com.feeyo.prophet.pojo.action.financial.FinancialReportReq;
import com.feeyo.prophet.pojo.action.fleet.FleetAttributesBody;
import com.feeyo.prophet.pojo.action.fleet.FleetAttributesEventBody;
import com.feeyo.prophet.pojo.action.flight.AirportIataReq;
import com.feeyo.prophet.pojo.action.rank.AirportRankReq;
import com.feeyo.prophet.pojo.action.rank.FltVolReqBody;
import com.feeyo.prophet.pojo.action.rank.MhkbPairReq;
import com.feeyo.prophet.pojo.action.rcdi.BrainAnswerResp;
import com.feeyo.prophet.pojo.action.rcdi.RcdiBeforeReq;
import com.feeyo.prophet.pojo.enums.ContentTypeEnum;
import com.feeyo.prophet.pojo.enums.UsefulnesEnum;
import com.feeyo.prophet.pojo.vo.AnswerVO;
import com.feeyo.prophet.service.RiakService;
import com.google.common.collect.Maps;
import com.opencsv.CSVWriter;
import io.micrometer.common.util.StringUtils;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.io.*;
import java.util.Iterator;

import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.*;
import java.util.zip.GZIPInputStream;

/**
 * 数据库交互action 配套util
 *
 * @author lihu
 */
@Slf4j
@Component
public class DbSearchActionUtils {

    private static RiakService riakService;
    private static WebClient webClient;
    private static final long LAST_AGE = ConstantUtils.FILE_CLEAN_AGE;

    @Value("${riak.bucket}")
    private String bucket;

    @Autowired
    private WebClient autowiredWebClient;

    @Autowired
    private RiakService riakServiceClient;

    @PostConstruct
    public void init() {
        webClient = autowiredWebClient;
        riakService = riakServiceClient;
    }

    /**
     * 定时任务，每天凌晨1点执行一次, 删除历史文档
     */
    @Scheduled(cron = "0 0 1 * * ?")
    public void cleanOldFiles() {
        try {
            riakService.deleteExpiration(bucket, LAST_AGE);
        } catch (Exception e) {
            log.error("cleanOldFiles error", e);
        }
    }

    /**
     * 查询节假日
     *
     * @param year 日期
     */
    public static String getHoliday(String year) {
        log.debug("getHoliday req:{}", JSON.toJSON(year));
        String respTxt;
        try {
            respTxt = webClient.method(HttpMethod.GET)
                    .uri(uriBuilder -> uriBuilder
                            .scheme("https")
                            .host("timor.tech")
                            .path("/api/holiday/year/" + year)
                            .build())
                    .retrieve()
                    .bodyToMono(String.class)
                    .block();
            log.debug("getHoliday resp:{}", respTxt);
        } catch (Exception e) {
            log.error("getHoliday error->{}", JSON.toJSON(year), e);
            throw e;
        }
        return respTxt;
    }

    /**
     * 数据存储成连接
     */
    public static String generateUrl(String baseUrl, String bucket, Object dataObject) throws Exception {
        String jsonString = JSON.toJSONString(dataObject);
        // 示例 JSON 数据
        String fileName = Md5Util.generate()+"."+ConstantUtils.CONTENT_TYPE_CSV;
        convertJsonToCsvAndStoreInRiak(jsonString, bucket, fileName);
        // 将 result 作为一个 HTML 文档保存
        String fileUrl = baseUrl + "/" + fileName;
        log.info("riak数据存储url,{}", fileUrl);
        return fileUrl;
    }

    public static void convertJsonToCsvAndStoreInRiak(String jsonData, String bucket, String fileName) throws Exception {
        ObjectMapper objectMapper = new ObjectMapper();
        JsonNode rootNode = objectMapper.readTree(jsonData.getBytes(StandardCharsets.UTF_8));
        // 假设JSON是一个数组，并且第一个对象包含所有可能的字段名
        JsonNode firstNode = !rootNode.isEmpty() ? rootNode.get(0) : null;
        if (firstNode == null) {
            throw new IllegalArgumentException("JSON array is empty or null");
        }
        Iterator<String> fieldNames = firstNode.fieldNames();
        // 计算字段名的数量
        int fieldCount = 0;
        while (fieldNames.hasNext()) {
            fieldNames.next();
            fieldCount++;
        }
        fieldNames = firstNode.fieldNames();
        // 创建头部数组
        String[] header = new String[fieldCount];
        int i = 0;
        while (fieldNames.hasNext()) {
            header[i++] = fieldNames.next();
        }
        // 使用 ByteArrayOutputStream 写入 CSV 数据到内存中
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        try (CSVWriter csvWriter = new CSVWriter(new OutputStreamWriter(byteArrayOutputStream, StandardCharsets.UTF_8))) {
            // 写入CSV标题行
            csvWriter.writeNext(header);
            // 写入CSV数据行
            for (JsonNode node : rootNode) {
                String[] row = new String[header.length];
                for (int j = 0; j < header.length; j++) {
                    String field = header[j];
                    // 获取字段值，如果字段不存在则使用空字符串
                    row[j] = node.has(field) ? node.get(field).asText() : "";
                }
                csvWriter.writeNext(row);
            }
        }
        // 将 CSV 数据存储到 Riak 中
        byte[] csvData = byteArrayOutputStream.toByteArray();
        riakSave(bucket, fileName, csvData, null);
    }

    /**
     * 指定类型的二进制文件存储
     * @param bucket
     * @param fileName
     * @param value
     * @param contentType
     * @throws Exception
     */
    public static void riakSave(String bucket, String fileName, byte[] value, String contentType) throws Exception {
        riakService.set(bucket, fileName, value, contentType);
    }

    /**
     * 文件预览
     * @param bucket
     * @param fileName
     * @return
     */
    public static String riakPreview(String bucket, String fileName) {
        return riakService.preview(bucket, fileName);
    }

    public static void inputAndOutputLog(String method, Object input, String output) {
        if(ObjectUtils.isNotEmpty(input)){
            log.info("\nMethod:{} Input:{}\n", method, JSON.toJSON(input));
        }
        if(StringUtils.isNotEmpty(output)){
            log.info("\nMethod Result:{}\n", output);
        }
    }

    /**
     * 航班查询
     */
    public static String queryFlightSchedule(String host, String fnum, String forg, String fdst, String flightDate) {
        MultiValueMap<String, String> queryMap = new LinkedMultiValueMap<>();
        if (StringUtils.isNotEmpty(forg)) {
            queryMap.add("forg", forg);
        }
        if (StringUtils.isNotEmpty(fdst)) {
            queryMap.add("fdst", fdst);
        }
        if (StringUtils.isNotEmpty(fnum)) {
            queryMap.add("fnum", fnum);
        }
        queryMap.add("fdepdate", flightDate);
        queryMap.add("noCancel", "true");
        inputAndOutputLog("queryFlightSchedule", queryMap, null);
        String respTxt;
        try {
            respTxt = webClient.method(HttpMethod.GET)
                    .uri(uriBuilder -> uriBuilder
                            .scheme("http")
                            .host(host)
                            .path("/api/fs/cal/query")
                            // 添加查询参数
                            .queryParams(queryMap)
                            .build())
                    .headers(headers -> headers.setAll(headerParams()))
                    .retrieve()
                    .bodyToMono(String.class)
                    .block();
            inputAndOutputLog("queryFlightSchedule", null, respTxt);
        } catch (Exception e) {
            log.error("queryFlightSchedule error->{}", JSON.toJSON(queryMap), e);
            throw e;
        }
        return respTxt;
    }

    /**
     * tradition：0:国内-国内(有时直接简述为国内),1:国内-地区(有时直接简述为地区),2:国内-国际(有时直接简述为国际),3:地区-地区,4:地区-国际,5:国际-国际'
     * customized：0：国内-国内，1：国内-国际，2：国内-地区，3：地区-国际，4；国际-国际，默认为 0
     */
    private static List<Integer> FlightDynamicFcategoryCovert(List<Integer> fcategoryList) {
        Set<Integer> fcategorySet = new HashSet<>();
        for (Integer fcategory : fcategoryList) {
            if(fcategory == 0){fcategorySet.add(0);}
            if(fcategory == 1){fcategorySet.add(2);}
            if(fcategory == 2){fcategorySet.add(1);}
            if(fcategory == 3){fcategorySet.add(0);}
            if(fcategory == 4){fcategorySet.add(3);}
            if(fcategory == 5){fcategorySet.add(4);}
        }
        return fcategorySet.stream().toList();
    }

    /**
     * 查询航班动态数据
     * <a href="http://yapi.feeyo.com/project/77/interface/api/493">获取航班动态信息</a>
     */
    public static String queryFlightDynamicData(String host, int port, String flightDate, String startTime, String endTime, String fservice, List<Integer> fcategory,
                                                String fnum, String anum, String depCode, String arrCode, String airline) throws Exception {
        FlightDynamicReq pair = new FlightDynamicReq();
        pair.setFlightDate(flightDate);
        if (StringUtils.isNotEmpty(startTime)) {
            pair.setStartTime(startTime);
        }
        if (StringUtils.isNotEmpty(endTime)) {
            pair.setEndTime(endTime);
        }
        if (StringUtils.isNotEmpty(fservice)) {
            pair.setFservice(new ArrayList<>(List.of(fservice.toLowerCase())));
        }
        //这是一个特殊的接口 fcategory 于其他fcatrgory不对应，需要转化
        if (CollectionUtils.isNotEmpty(fcategory)) {
            pair.setFcategory(FlightDynamicFcategoryCovert(fcategory));
        }
        if (StringUtils.isNotEmpty(fnum)) {
            pair.setFnum(fnum);
        }
        if (StringUtils.isNotEmpty(anum)) {
            pair.setAnum(anum);
        }
        if (StringUtils.isNotEmpty(depCode)) {
            pair.setDep(depCode);
        }
        if (StringUtils.isNotEmpty(arrCode)) {
            pair.setArr(arrCode);
        }
        if (StringUtils.isNotEmpty(airline)) {
            pair.setAirline(airline);
        }
        inputAndOutputLog("queryFlightDynamicData", pair, null);
        String respTxt;
        try {
            URI uri = new URI("http", null, host, port, "/dynamic/api/v1/flight/dynamic/data", null, null);
            respTxt = webClient.method(HttpMethod.POST)
                    .uri(uri)
                    .headers(headers -> headers.setAll(newHeader()))
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(Mono.just(pair), FlightDynamicReq.class)
                    .retrieve()
                    .bodyToMono(String.class)
                    .block();
            inputAndOutputLog("queryFlightDynamicData", null, respTxt);
        } catch (WebClientResponseException e) {
            log.error("queryFlightDynamicData error (status code: {})", e.getStatusCode(), e);
            throw e;
        } catch (Exception e) {
            log.error("queryFlightDynamicData error->{}", pair, e);
            throw e;
        }
        JSONObject json = JSONObject.parseObject(respTxt);
        JSONObject data = json.getJSONObject("data");
        if (null == data) {
            return "";
        }
        JSONArray row = data.getJSONArray("rows");
        if (null == row) {
            return "";
        }
        return respTxt;
    }

    /**
     * 点对点,点对线, 包含国对国,非明细数据查询.
     * <p/>
     * <a href="http://yapi.feeyo.com/project/78/interface/api/699">获取民航看板航段航司总数信息</a>
     */
    public static String queryDepArr(String host, int port,
                                     String startDate, String endDate,
                                     List<String> forgList, List<String> fdstList,
                                     String forgType,
                                     String fdstType,
                                     String fcategory, String viewType, String fservice, List<String> airlines) throws Exception {
        MhkbPairReq pair = new MhkbPairReq();
        pair.setSdepdateStart(startDate);
        pair.setSdepdateEnd(endDate);
        pair.setOrgs(forgList);
        pair.setDsts(fdstList);
        pair.setViewType(viewType);
        pair.setForgType(forgType);
        pair.setFdstType(fdstType);
        if (StringUtils.isNotEmpty(fservice)) {
            pair.setFservice(fservice);
        }
        if (StringUtils.isNotEmpty(fcategory)) {
            pair.setFcate(fcategory);
        }
        if (CollectionUtils.isNotEmpty(airlines)) {
            pair.setAirlines(airlines);
        }
        inputAndOutputLog("queryDepArr", pair, null);
        String respTxt;
        try {
            URI uri = new URI("http", null, host, port, "/calc/api/v1/mhkb/getPairAirlineCount", null, null);
            respTxt = webClient.method(HttpMethod.POST)
                    .uri(uri)
                    .headers(headers -> headers.setAll(newHeader()))
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(Mono.just(pair), MhkbPairReq.class)
                    .retrieve()
                    .bodyToMono(String.class)
                    .block();
            inputAndOutputLog("queryDepArr", null, respTxt);
        } catch (Exception e) {
            log.error("queryDepArr error->{}", pair, e);
            throw e;
        }
        return respTxt;
    }

    /**
     * fcategory 查询
     */
    public static InterfaceQueryResult queryFcategory(String host, int port, String startDate, String endDate, String fcategory, String fservice) throws Exception {
        InterfaceQueryResult result = new InterfaceQueryResult();
        result.setStatusCode(HttpStatus.OK.value());
        FltVolReqBody fltVolReqBody = new FltVolReqBody();
        fltVolReqBody.setStartDate(startDate);
        fltVolReqBody.setEndDate(endDate);
        fltVolReqBody.setFcategory(fcategory);
        if (StringUtils.isNotEmpty(fservice)) {
            fltVolReqBody.setFservice(fservice);
        }
        String respTxt = "";
        inputAndOutputLog("queryFcategory", fltVolReqBody, null);
        try {
            URI uri = new URI("http", null, host, port, "/calc/api/v1/statistics/flight/volume/fcategory", null, null);
            respTxt = webClient.method(HttpMethod.POST)
                    .uri(uri)
                    .headers(headers -> headers.setAll(DbSearchActionUtils.newHeader()))
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(Mono.just(fltVolReqBody), FltVolReqBody.class)
                    .retrieve()
                    .bodyToMono(String.class)
                    .block();
            inputAndOutputLog("queryFcategory", null, respTxt);
        } catch (WebClientResponseException e) {
            log.error("searchPriceByDate error (status code: {})", e.getStatusCode(), e);
            result.setStatusCode(e.getStatusCode().value());
        } catch (Exception e) {
            log.error("queryFcategory error ->{}", fltVolReqBody, e);
        }
        result.setResponseText(respTxt);
        return result;
    }

    /**
     * 航司
     */
    public static String queryAirline(String host, int port, String startDate, String endDate, String airline, String fcategory, String fservice) throws Exception {
        FltVolReqBody fltVolReqBody = new FltVolReqBody();
        fltVolReqBody.setStartDate(startDate);
        fltVolReqBody.setEndDate(endDate);
        fltVolReqBody.setAirline(airline);
        if (StringUtils.isNotEmpty(fcategory)) {
            fltVolReqBody.setFcategory(fcategory);
        }
        if (StringUtils.isNotEmpty(fservice)) {
            fltVolReqBody.setFservice(fservice);
        }
        String respTxt;
        inputAndOutputLog("queryAirline", fltVolReqBody, null);
        try {
            URI uri = new URI("http", null, host, port, "/calc/api/v1/statistics/flight/volume/airline", null, null);
            respTxt = webClient.method(HttpMethod.POST)
                    .uri(uri)
                    .headers(headers -> headers.setAll(DbSearchActionUtils.newHeader()))
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(Mono.just(fltVolReqBody), FltVolReqBody.class)
                    .retrieve()
                    .bodyToMono(String.class)
                    .block();
            inputAndOutputLog("queryAirline", null, respTxt);
        } catch (Exception e) {
            log.error("queryAirline error->{}", fltVolReqBody, e);
            throw e;
        }
        return respTxt;
    }

    /**
     * 航司排名
     */
    public static String queryAirlineRank(String host, int port, String startDate, String endDate, String airline) throws Exception {
        FltVolReqBody fltVolReqBody = new FltVolReqBody();
        fltVolReqBody.setStartDate(startDate);
        fltVolReqBody.setEndDate(endDate);
        fltVolReqBody.setAirline(airline);
        String respTxt;
        inputAndOutputLog("queryAirlineRank", fltVolReqBody, null);
        try {
            URI uri = new URI("http", null, host, port, "/calc/api/v1/statistics/flight/volume/airline/rank", null, null);
            respTxt = webClient.method(HttpMethod.POST)
                    .uri(uri)
                    .headers(headers -> headers.setAll(DbSearchActionUtils.newHeader()))
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(Mono.just(fltVolReqBody), FltVolReqBody.class)
                    .retrieve()
                    .bodyToMono(String.class)
                    .block();
            inputAndOutputLog("queryAirlineRank", null, respTxt);
        } catch (Exception e) {
            log.error("queryAirlineRank error ->{}", fltVolReqBody, e);
            throw e;
        }
        return respTxt;
    }


    public static String queryAirportIata(String host, int port, String cityIata) throws Exception {
        AirportIataReq iataReq = new AirportIataReq();
        iataReq.setCityiata(cityIata);
        String respTxt;
        inputAndOutputLog("queryAirportIata", iataReq, null);
        try {
            URI uri = new URI("http", null, host, port, "/base/api/v1/airport/data", null, null);
            respTxt = webClient.method(HttpMethod.POST)
                    .uri(uri)
                    .headers(headers -> headers.setAll(DbSearchActionUtils.newHeader()))
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(Mono.just(iataReq), FltVolReqBody.class)
                    .retrieve()
                    .bodyToMono(String.class)
                    .block();
            inputAndOutputLog("queryAirportIata", null, respTxt);
        } catch (Exception e) {
            log.error("queryAirportIata error->{}", iataReq, e);
            throw e;
        }
        return respTxt;
    }

    /**
     * 指定机场排名
     */
    public static String specifyAirportRank(String host, int port, String startDate, String endDate, String airport, String category) throws Exception {
        AirportRankReq airportRankReq = new AirportRankReq();
        airportRankReq.setStartDate(startDate);
        airportRankReq.setEndDate(endDate);
        airportRankReq.setAirport(airport);
        int currentCategory = 0;
        if (StringUtils.isNotEmpty(category) && "国际".equals(category)) {
            currentCategory = 1;
        }
        airportRankReq.setCategory(currentCategory);
        String respTxt;
        inputAndOutputLog("specifyAirportRank", airportRankReq, null);
        try {
            URI uri = new URI("http", null, host, port, "/calc/api/v1/throughput/specifyAirportRank", null, null);
            respTxt = webClient.method(HttpMethod.POST)
                    .uri(uri)
                    .headers(headers -> headers.setAll(newHeader()))
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(Mono.just(airportRankReq), AirportRankReq.class)
                    .retrieve()
                    .bodyToMono(String.class)
                    .block();
            inputAndOutputLog("specifyAirportRank", null, respTxt);
        } catch (Exception e) {
            log.error("specifyAirportRank error ->{}", airportRankReq, e);
            throw e;
        }
        return respTxt;
    }

    /**
     * 全国范围机场排名
     */
    public static String domesticAirportsRank(String host, int port, String startDate, String endDate, int count) throws Exception {
        AirportRankReq airportRankReq = new AirportRankReq();
        airportRankReq.setStartDate(startDate);
        airportRankReq.setEndDate(endDate);
        airportRankReq.setCount(count);
        String respTxt;
        inputAndOutputLog("domesticAirportsRank", airportRankReq, null);
        try {
            URI uri = new URI("http", null, host, port, "/calc/api/v1/throughput/domesticAirportsRank", null, null);
            respTxt = webClient.method(HttpMethod.POST)
                    .uri(uri)
                    .headers(headers -> headers.setAll(newHeader()))
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(Mono.just(airportRankReq), AirportRankReq.class)
                    .retrieve()
                    .bodyToMono(String.class)
                    .block();
            inputAndOutputLog("domesticAirportsRank", null, respTxt);
        } catch (Exception e) {
            log.error("domesticAirportsRank error ->{}", airportRankReq, e);
            throw e;
        }
        return respTxt;
    }

    /**
     * 千万级以下机场排名
     */
    public static String getAirportsRankUnder10m(String host, int port, String startDate, String endDate, int count) {
        AirportRankReq airportRankReq = new AirportRankReq();
        airportRankReq.setStartDate(startDate);
        airportRankReq.setEndDate(endDate);
        airportRankReq.setCount(count);
        String respTxt = null;
        inputAndOutputLog("getAirportsRankUnder10m", airportRankReq, null);
        try {
            URI uri = new URI("http", null, host, port, "/calc/api/v1/throughput/getAirportsRankUnder10m", null, null);
            respTxt = webClient.method(HttpMethod.POST)
                    .uri(uri)
                    .headers(headers -> headers.setAll(newHeader()))
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(Mono.just(airportRankReq), AirportRankReq.class)
                    .retrieve()
                    .bodyToMono(String.class)
                    .block();
            inputAndOutputLog("getAirportsRankUnder10m", null, respTxt);
        } catch (Exception e) {
            log.error("getAirportsRankUnder10m error ->{}", airportRankReq, e);
        }
        return respTxt;
    }

    public static String getAirportsRankOver10m(String host, int port, String startDate, String endDate, int count) {
        AirportRankReq airportRankReq = new AirportRankReq();
        airportRankReq.setStartDate(startDate);
        airportRankReq.setEndDate(endDate);
        airportRankReq.setCount(count);
        String respTxt = null;
        inputAndOutputLog("getAirportsRankOver10m", airportRankReq, null);
        try {
            URI uri = new URI("http", null, host, port, "/calc/api/v1/throughput/getAirportsRankOver10m", null, null);
            respTxt = webClient.method(HttpMethod.POST)
                    .uri(uri)
                    .headers(headers -> headers.setAll(newHeader()))
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(Mono.just(airportRankReq), AirportRankReq.class)
                    .retrieve()
                    .bodyToMono(String.class)
                    .block();
            inputAndOutputLog("getAirportsRankOver10m", null, respTxt);
        } catch (Exception e) {
            log.error("getAirportsRankOver10m error ->{}", airportRankReq, e);
        }
        return respTxt;
    }

    /**
     * 时刻协调机场排名
     */
    public static String getTimeAirportsRank(String host, int port, String startDate, String endDate, int count) {
        AirportRankReq airportRankReq = new AirportRankReq();
        airportRankReq.setStartDate(startDate);
        airportRankReq.setEndDate(endDate);
        airportRankReq.setCount(count);
        String respTxt = null;
        inputAndOutputLog("getTimeAirportsRank", airportRankReq, null);
        try {
            URI uri = new URI("http", null, host, port, "/calc/api/v1/throughput/getTimeAirportsRank", null, null);
            respTxt = webClient.method(HttpMethod.POST)
                    .uri(uri)
                    .headers(headers -> headers.setAll(newHeader()))
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(Mono.just(airportRankReq), AirportRankReq.class)
                    .retrieve()
                    .bodyToMono(String.class)
                    .block();
            inputAndOutputLog("getTimeAirportsRank", null, respTxt);
        } catch (Exception e) {
            log.error("getTimeAirportsRank error ->{}", airportRankReq, e);
        }
        return respTxt;
    }

    /**
     * 机场吞吐量分级查询
     */
    public static String getAirports10m(String host, int port, int airPort, int count) {
        AirportRankReq airportRankReq = new AirportRankReq();
        airportRankReq.setAirportType(airPort);
        airportRankReq.setCount(count);
        inputAndOutputLog("getAirports10m", airportRankReq, null);
        String respTxt = null;
        try {
            URI uri = new URI("http", null, host, port, "/calc/api/v1/throughput/getAirports10m", null, null);
            respTxt = webClient.method(HttpMethod.POST)
                    .uri(uri)
                    .headers(headers -> headers.setAll(newHeader()))
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(Mono.just(airportRankReq), AirportRankReq.class)
                    .retrieve()
                    .bodyToMono(String.class)
                    .block();
            inputAndOutputLog("getAirports10m", null, respTxt);
        } catch (Exception e) {
            log.error("getAirports10m error ->{}", airportRankReq, e);
        }
        return respTxt;
    }

    /**
     * 查询机场金融信息
     *
     * @param host    host
     * @param port    端口
     * @param airport 机场
     * @param year    时间
     */
    public static String queryFinancialAirport(String host, Integer port, String airport, String airline, String year) throws Exception {
        URI uri = null;
        FinancialReportReq reqBody = new FinancialReportReq();
        reqBody.setYear(year);
        if (StringUtils.isNotEmpty(airline)) {
            reqBody.setAirline(airline);
            uri = new URI("http", null, host, port, "/calc/api/v1/financial/statistic/airline", null, null);
        }
        if (StringUtils.isNotEmpty(airport)) {
            reqBody.setAirport(airport);
            uri = new URI("http", null, host, port, "/calc/api/v1/financial/statistic/airport", null, null);
        }
        inputAndOutputLog("queryFinancialAirport", reqBody, null);
        String respTxt;
        assert uri != null;
        respTxt = webClient.method(HttpMethod.POST)
                .uri(uri)
                .headers(headers -> headers.setAll(DbSearchActionUtils.newHeader()))
                .contentType(MediaType.APPLICATION_JSON)
                .body(Mono.just(reqBody), FinancialReportReq.class)
                .retrieve()
                .bodyToMono(String.class)
                .block();
        inputAndOutputLog("queryFinancialAirport", null, respTxt);
        return respTxt;
    }

    /**
     * 查询运营机型
     */
    public static String queryOperatorAircraft(String host, Integer port, String operatorIata, String manager, String aircraftManufacturer, List<String> aircraftStatusList, List<String> aircraftTypeList) {
        FleetAttributesBody reqBody = new FleetAttributesBody();
        if (StringUtils.isNotEmpty(operatorIata)) {
            reqBody.setOperatorIata(operatorIata);
        }
        if (StringUtils.isNotEmpty(manager)) {
            reqBody.setManager(manager);
        }
        if (StringUtils.isNotEmpty(aircraftManufacturer)) {
            reqBody.setAircraftManufacturer(aircraftManufacturer);
        }
        if (CollectionUtils.isNotEmpty(aircraftTypeList)) {
            reqBody.setAircraftTypeList(aircraftTypeList);
        }
        if (CollectionUtils.isNotEmpty(aircraftStatusList)) {
            reqBody.setAircraftStatusList(aircraftStatusList);
        }
        inputAndOutputLog("queryOperatorAircraft", reqBody, null);
        String respTxt = null;
        try {
            URI uri = new URI("http", null, host, port, "/base/api/v1/fleet/data", null, null);
            respTxt = webClient.method(HttpMethod.POST)
                    .uri(uri)
                    .headers(headers -> headers.setAll(DbSearchActionUtils.newHeader()))
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(Mono.just(reqBody), FleetAttributesBody.class)
                    .retrieve()
                    .bodyToMono(String.class)
                    .block();
            inputAndOutputLog("queryOperatorAircraft", null, respTxt);
        } catch (Exception e) {
            log.error("DbSearchActionUtils queryOperatorAircraft error->{}", reqBody, e);
        }
        return respTxt;
    }

    /**
     * 查询机型事件
     *
     * @param host       IP
     * @param port       PORT
     * @param aircraftId 飞机ID
     */
    public static String queryOperatorEventAircraft(String host, Integer port, String aircraftId) {
        FleetAttributesEventBody reqBody = new FleetAttributesEventBody();
        if (StringUtils.isNotEmpty(aircraftId)) {
            reqBody.setAircraftId(aircraftId);
        }
        inputAndOutputLog("queryOperatorEventAircraft", reqBody, null);
        String respTxt = null;
        try {
            URI uri = new URI("http", null, host, port, "/base/api/v1/fleet/event/data", null, null);
            respTxt = webClient.method(HttpMethod.POST)
                    .uri(uri)
                    .headers(headers -> headers.setAll(DbSearchActionUtils.newHeader()))
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(Mono.just(reqBody), FleetAttributesEventBody.class)
                    .retrieve()
                    .bodyToMono(String.class)
                    .block();
            inputAndOutputLog("queryOperatorEventAircraft", null, respTxt);
        } catch (Exception e) {
            log.error("DbSearchActionUtils queryOperatorEventAircraft error->{}", reqBody, e);
        }
        return respTxt;
    }

    /**
     * 查询
     */
    public static String searchFeaturePriceByDate(String host, Integer port, String startDate, String endDate, String forg, String fdst, String fnum) {
        log.info("searchPriceByDate req: startDate={}, endDate={}, forg={}, fdst={}, fnum={}", startDate, endDate, forg, fdst, fnum);
        String respTxt = null;
        try {
            WebClient.RequestHeadersUriSpec<?> uriSpec = (WebClient.RequestHeadersUriSpec<?>) webClient.get()
                    .uri(uriBuilder -> {
                        uriBuilder
                                .scheme("http")
                                .host(host)
                                .port(port)
                                .path("/api/ff/matrix/lowest/price/checkpoint/query")
                                .queryParam("startDate", startDate)
                                .queryParam("endDate", endDate)
                                .queryParam("forg", forg)
                                .queryParam("fdst", fdst);
                        if (StringUtils.isNotEmpty(fnum)) {
                            uriBuilder.queryParam("fnum", fnum);
                        }
                        return uriBuilder.build();
                    });
            respTxt = uriSpec
                    .header("Accept-Encoding", "gzip")
                    .retrieve()
                    .bodyToMono(byte[].class)
                    .map(DbSearchActionUtils::decompressGzip)
                    .block();
            inputAndOutputLog("searchPriceByDate", null, respTxt);
        } catch (Exception e) {
            log.error("searchPriceByDate error -> {}", fdst, e);
        }
        return respTxt;
    }

    public static String searchHistoryPriceByDate(String host, Integer port, String startDate, String endDate, String forg, String fdst, String fnum, String fdeptDate) {
        log.info("searchHistoryPriceByDate req: startDate={}, endDate={}, forg={}, fdst={}, fnum={}", startDate, endDate, forg, fdst, fnum);
        String respTxt = null;
        try {
            WebClient.RequestHeadersUriSpec<?> uriSpec = (WebClient.RequestHeadersUriSpec<?>) webClient.get()
                    .uri(uriBuilder -> {
                        uriBuilder
                                .scheme("http")
                                .host(host)
                                .port(port)
                                .path("/api/ff/matrix/lowest/price/history/query")
                                .queryParam("startDate", startDate)
                                .queryParam("endDate", endDate)
                                .queryParam("forg", forg)
                                .queryParam("fdst", fdst)
                                .queryParam("fdepdate", fdeptDate);
                        if (StringUtils.isNotEmpty(fnum)) {
                            uriBuilder.queryParam("fnum", fnum);
                        }
                        return uriBuilder.build();
                    });
            respTxt = uriSpec
                    .header("Accept-Encoding", "gzip")
                    .retrieve()
                    .bodyToMono(byte[].class)
                    .map(DbSearchActionUtils::decompressGzip)
                    .block();
            inputAndOutputLog("searchPriceByDate", null, respTxt);
        } catch (Exception e) {
            log.error("searchPriceByDate error -> {}", fdst, e);
        }
        return respTxt;
    }



    private static String decompressGzip(byte[] compressed) {
        try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(compressed);
             GZIPInputStream gzipInputStream = new GZIPInputStream(byteArrayInputStream)) {
            byte[] buffer = new byte[1024];
            StringBuilder outStr = new StringBuilder();
            int len;
            while ((len = gzipInputStream.read(buffer)) != -1) {
                outStr.append(new String(buffer, 0, len, StandardCharsets.UTF_8));
            }
            return outStr.toString();
        } catch (IOException e) {
            throw new RuntimeException("Failed to decompress gzip response", e);
        }
    }

    /**
     * @param question 具体查询的问题
     */
    public static String getRcdiBefore(String host, int port, String question, String chatId) {
        RcdiBeforeReq pair = new RcdiBeforeReq();
        pair.setQuestion(question);
        pair.setChatId(chatId);
        inputAndOutputLog("searchPriceByDate", pair, null);
        String respTxt = null;
        try {
            URI uri = new URI("http", null, host, port, "/web/handle/chat/sse/before", null, null);
            respTxt = webClient.method(HttpMethod.POST)
                    .uri(uri)
                    .headers(headers -> headers.setAll(basicNewHeader()))
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(Mono.just(pair), RcdiBeforeReq.class)
                    .retrieve()
                    .bodyToMono(String.class)
                    .block();
            inputAndOutputLog("searchPriceByDate", null, respTxt);
        } catch (Exception e) {
            log.error("RcdiService getRcdiBefore error->{}", pair, e);
        }
        return respTxt;
    }

    /**
     * rcdi sse 数据提交
     */
    public static void getRcdiSse(String host, int port, String id, String rcdiId, String lastId, SseEmitter emitter) {
        String url = String.format("http://%s:%d/web/handle/chat/sse?x=1&id=%s", host, port, rcdiId);
        log.info("Connecting to SSE endpoint: {}", url);
        webClient.get()
                .uri(url)
                .header("Accept", "text/event-stream")
                .header("Cache-Control", "no-cache")
                .exchangeToFlux(response -> {
                    if (response.statusCode().is2xxSuccessful()) {
                        return response.bodyToFlux(String.class);
                    } else {
                        return Flux.error(new RuntimeException("Failed to connect to SSE endpoint"));
                    }
                })
                .doOnError(error -> log.error("Error receiving SSE data", error))
                .subscribe(
                        data -> {
                            if ("done".equals(data)) {
                                // Handle completion
                                String answerVO = JsonUtil.marshalToString(BrainAnswerResp.getBrainAnswerStep(ConstantUtils.COMMON_INT_TEN, "done"));
                                sseResponse(id, answerVO, lastId, emitter);
                                log.info("SSE stream completed.");
                                emitter.complete();
                            } else {
                                // Process the response
                                sseResponse(id, data, lastId,  emitter);
                            }
                        },
                        error -> log.error("Error receiving SSE data", error),
                        () -> log.info("SSE stream closed.")
                );
    }

    /**
     * 处理响应, 会有重连，增加ignore
     */
    public static void sseResponse(String id, String respData, String lastId, SseEmitter emitter) {
        log.debug("Processing response: {}", respData);
        try {
            BrainAnswerResp brainAnswerResp = JSON.parseObject(respData, BrainAnswerResp.class);
            AnswerVO answerVO = new AnswerVO();
            answerVO.setId(id);
            answerVO.setLastId(lastId);
            answerVO.setContentType(ContentTypeEnum.TEXT.getCode().toUpperCase());
            answerVO.setCreateTime(LocalDateTime.now());
            answerVO.setUsefulness(UsefulnesEnum.NON.getCode());
            answerVO.setContent(JsonUtil.marshalToString(brainAnswerResp));
            emitter.send(SseEmitter.event().data(answerVO));
        } catch (Exception e) {
            log.error("sseResponse error -> {}", e.getMessage());
        }
    }
    
    public static void sseResponse(String result,  SseEmitter emitter) {
        try {
            emitter.send(SseEmitter.event().data(result));
        } catch (Exception e) {
            log.error("sseResponse Result Error -> {}", e.getMessage());
        }
    }

    public static Map<String, String> basicNewHeader() {
        return Maps.newHashMap();
    }

    /**
     * 航班token
     */
    public static Map<String, String> headerParams() {
        return new HashMap<>(8) {{
            put("token", ConstantUtils.SCHEDULE_TOKEN);
            put("timestamp", ConstantUtils.SCHEDULE_TIMESTAMP);
            put("appId", ConstantUtils.SCHEDULE_APP_ID);
        }};
    }

    /**
     * 基础数据token
     */
    public static Map<String, String> newHeader() {
        String appId = ConstantUtils.BASIC_APP_ID;
        String security = ConstantUtils.BASIC_SECURITY;
        Map<String, String> headerMap = Maps.newHashMap();
        Long timestamp = new Date().getTime() / 1000;
        String token = Md5Util.md5_32(Md5Util.md5_32(security) + timestamp);
        headerMap.put("appId", appId);
        headerMap.put("timestamp", String.valueOf(timestamp));
        headerMap.put("token", token);
        log.info("headerMap:{}", headerMap);
        return headerMap;
    }

}
